package cn.azzhu.day03.window;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * 使用EventTime，划分滚动窗口
 * Flink划分窗口，窗口的时间精确到毫秒，窗口是前闭后开的
 * 如果使用的是并行的Source，例如KafkaSource，创建Kafka的Topic时有多个分区
 * 每一个Source的分区都要满足触发的条件，整个窗口才会被触发
 *
 * @author azzhu
 * @create 2020-09-19 01:11:20
 */
public class SocketSourceEventTimeSlidingWindow {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //设置eventtime作为时间标准
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //1000,spark,3
        //仅仅提取时间字段，不会改变数据的样式
        final SingleOutputStreamOperator<String> lines = env.socketTextStream("hadoop105", 9999)
                .assignTimestampsAndWatermarks(
                        new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(0)) {
                            //将数据中的时间字段提取出来，然后转为long类型
                            @Override
                            public long extractTimestamp(String line) {
                                final String[] fields = line.split(",");
                                return Long.parseLong(fields[0]);
                            }
                        });

        final SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = lines.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String line) throws Exception {
                final String[] fields = line.split(",");
                return Tuple2.of(fields[1], Integer.parseInt(fields[2]));
            }
        });

        //先分组，再划分窗口
        final KeyedStream<Tuple2<String, Integer>, Tuple> keyed = wordAndCount.keyBy(0);

        final WindowedStream<Tuple2<String, Integer>, Tuple, TimeWindow> window =
                keyed.window(SlidingEventTimeWindows.of(Time.seconds(5), Time.seconds(2)));

        //在窗口中聚合
        final SingleOutputStreamOperator<Tuple2<String, Integer>> summed = window.sum(1);
        summed.print();

        env.execute("SessionWindow");
    }
}
